package org.ghost.akka.cluster.master;

import akka.actor.AbstractActor;
import akka.actor.ActorRef;
import akka.actor.Terminated;
import akka.event.Logging;
import akka.event.LoggingAdapter;
import org.ghost.akka.cluster.Constants;
import org.ghost.akka.common.command.BaseCommand;
import org.ghost.akka.common.em.ResponseEnum;
import org.ghost.akka.common.model.JobResult;

import java.util.ArrayList;
import java.util.List;

/**
 * 任务分发器
 */
public class TaskDispatchActor extends AbstractActor {
    private LoggingAdapter log = Logging.getLogger(getContext().system(), this);
    private List<ActorRef> backendList = new ArrayList<>();
    private int jobCounter = 0;

    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(BaseCommand.class,
                        cmd -> backendList.isEmpty(),
                        cmd -> {
                            sender().tell(new JobResult(ResponseEnum.SERVICE_UNAVAILABLE), sender());
                        })
                .matchEquals(Constants.BACKEND_REGISTRATION, message -> {
                    getContext().watch(sender());
                    backendList.add(sender());
                })
                .match(Terminated.class, terminated -> {
                    backendList.remove(terminated.getActor());
                })
                .match(BaseCommand.class,
                        cmd -> {
                            log.info("message arrive task dispacher : " + cmd.getJob().getTaskId());
                            jobCounter++;
                            backendList.get(jobCounter % backendList.size()).forward(cmd, getContext());
                            log.info("current backend service count {}", backendList.size());
                        }).build();
    }
}
